package cn._51doit.day10;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @create: 2021-10-28 15:06
 * @author: 今晚打脑斧先森
 * @program: Demo02
 * @Description:
 * 练习 Broadcast State ，广播状态
 **/
public class Demo02 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> lines1 = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Tuple3<String, String, String>> tpStream = lines1.map(new MapFunction<String, Tuple3<String, String, String>>() {
            @Override
            public Tuple3<String, String, String> map(String value) throws Exception {
                //商品编号，商品名称，商品状态
                String[] split = value.split(",");
                return Tuple3.of(split[0], split[1], split[2]);
            }
        });
        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<>("商品，状态", String.class, String.class);
        BroadcastStream<Tuple3<String, String, String>> broadcastStream = tpStream.broadcast(mapStateDescriptor);

        DataStreamSource<String> lines2 = env.socketTextStream("doit01", 9999);
        SingleOutputStreamOperator<Tuple3<String, String, Integer>> tpStream2 = lines2.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> map(String value) throws Exception {
                //商品编号，购买人名字，购买个数
                String[] split = value.split(",");
                return Tuple3.of(split[0], split[1], Integer.parseInt(split[2]));
            }
        });
        tpStream2.connect(broadcastStream)
                .process(new BroadcastProcessFunction<Tuple3<String, String, Integer>, Tuple3<String, String, String>, Tuple4<String,String,String,Integer>>() {
                    @Override
                    public void processElement(Tuple3<String, String, Integer> value, ReadOnlyContext ctx, Collector<Tuple4<String, String, String, Integer>> out) throws Exception {
                        ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        String s = broadcastState.get(value.f0);
                        //商品编号，商品名称，购买人姓名，购买个数
                        out.collect(Tuple4.of(value.f0,s,value.f1,value.f2));
                    }

                    @Override
                    public void processBroadcastElement(Tuple3<String, String, String> value, Context ctx, Collector<Tuple4<String, String, String, Integer>> out) throws Exception {
                        BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        if ("delete".equals(value.f2)){
                            broadcastState.remove(value.f0);
                        }else {
                            broadcastState.put(value.f0,value.f1);
                        }
                    }
                }).print();
        env.execute();
    }
}
